﻿using KafkaUtil.Consumers;
using KafkaUtil.Integration;
using KafkaUtil.Interface;
using KafkaUtil.Producers;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.DependencyInjection.Extensions;
using Microsoft.Extensions.Hosting;
using Microsoft.Extensions.Options;
using System;
using System.Linq;
using System.Threading.Tasks;

namespace KafkaUtil
{
    public static class KafkaExtensions
    {
        /// <summary>
        /// 注册添加一个发送者
        /// </summary>
        /// <param name="services"></param>
        /// <param name="configure"></param>
        /// <returns></returns>
        public static IServiceCollection AddKafkaProducer(this IServiceCollection services, Action<KafkaProducerOptions> configure)
        {
            return services.AddKafkaProducer(Options.DefaultName, configure);
        }
        /// <summary>
        /// 注册添加一个发送者
        /// </summary>
        /// <param name="services"></param>
        /// <param name="configure"></param>
        /// <returns></returns>
        public static IServiceCollection AddKafkaProducer(this IServiceCollection services, string name, Action<KafkaProducerOptions> configure)
        {
            services.TryAddSingleton<IKafkaProducerFactory, KafkaProducerFactory>();
            services.Configure(name, configure);
            return services;
        }

        /// <summary>
        /// 注册一个消费者
        /// </summary>
        /// <param name="services"></param>
        /// <returns></returns>
        public static IServiceCollection AddKafkaConsumerCore(this IServiceCollection services)
        {
            if (!services.Any(f => f.ImplementationType == typeof(DefaultConsumerHostedService)))
            {
                services.AddSingleton<IHostedService, DefaultConsumerHostedService>();
            }
            return services;
        }
        /// <summary>
        /// 注册一个消费者
        /// </summary>
        /// <param name="services"></param>
        /// <param name="serviceType"></param>
        /// <returns></returns>
        public static IServiceCollection AddKafkaConsumer(this IServiceCollection services, Type serviceType)
        {
            if (!typeof(IKafkaConsumerProvider).IsAssignableFrom(serviceType) || !serviceType.IsClass || serviceType.IsAbstract)
            {
                throw new InvalidOperationException($"serviceType must be implement {nameof(IKafkaConsumerProvider)} and not abstract class ");
            }

            services.AddKafkaConsumerCore();
            services.AddSingleton(serviceType);
            return services;
        }
        /// <summary>
        /// 注册一个消费者
        /// </summary>
        /// <typeparam name="T"></typeparam>
        /// <param name="services"></param>
        /// <returns></returns>
        public static IServiceCollection AddKafkaConsumer<T>(this IServiceCollection services) where T : class, IKafkaConsumerProvider
        {
            return services.AddKafkaConsumer(typeof(T));
        }
        /// <summary>
        /// 注册一个消费者
        /// </summary>
        /// <param name="services"></param>
        /// <param name="kafkaConsumerProvider"></param>
        /// <returns></returns>
        public static IServiceCollection AddKafkaConsumer(this IServiceCollection services, IKafkaConsumerProvider kafkaConsumerProvider)
        {
            services.AddKafkaConsumerCore();
            services.AddSingleton(kafkaConsumerProvider);
            return services;
        }

        /// <summary>
        /// 注册消费者服务，获取消费创建对象
        /// </summary>
        /// <param name="services"></param>
        /// <param name="configure"></param>
        /// <returns></returns>
        public static IKafkaConsumerBuilder AddKafkaConsumer(this IServiceCollection services, Action<KafkaConsumerOptions> configure)
        {
            KafkaConsumerOptions kafkaConsumerOptions = new KafkaConsumerOptions();
            configure?.Invoke(kafkaConsumerOptions);
            return services.AddKafkaConsumer(kafkaConsumerOptions);
        }
        /// <summary>
        /// 注册消费者服务，获取消费创建对象
        /// </summary>
        /// <param name="services"></param>
        /// <param name="kafkaConsumerOptions"></param>
        /// <returns></returns>
        public static IKafkaConsumerBuilder AddKafkaConsumer(this IServiceCollection services, KafkaConsumerOptions kafkaConsumerOptions)
        {
            services.AddKafkaConsumerCore();
            return new KafkaConsumerBuilder(services, kafkaConsumerOptions);
        }
        /// <summary>
        /// 添加监听
        /// </summary>
        /// <param name="builder"></param>
        /// <param name="onMessageRecieved"></param>
        /// <returns></returns>
        public static IKafkaConsumerBuilder AddListener(this IKafkaConsumerBuilder builder, Action<RecieveResult> onMessageRecieved)
        {
            return builder.AddListener((_, result) =>
            {
                onMessageRecieved?.Invoke(result);
            });
        }
        /// <summary>
        /// 添加自定义监听
        /// </summary>
        /// <param name="builder"></param>
        /// <param name="listenerType"></param>
        /// <returns></returns>
        public static IKafkaConsumerBuilder AddListener(this IKafkaConsumerBuilder builder, Type listenerType)
        {
            if (!typeof(IKafkaConsumerListener).IsAssignableFrom(listenerType) || !listenerType.IsClass || listenerType.IsAbstract)
            {
                throw new ArgumentException($"the listener type must be implement IKafkaConsumerListener and none abstract class", "listenerType");
            }

            builder.Services.AddTransient(listenerType);
            return builder.AddListener((serviceProvider, result) =>
            {
                var listenner = serviceProvider.GetService(listenerType) as IKafkaConsumerListener;
                listenner.ConsumeAsync(result).Wait();
            });
        }
        /// <summary>
        /// 添加自定义监听
        /// </summary>
        /// <param name="builder"></param>
        /// <typeparam name="T"></typeparam>
        /// <returns></returns>
        public static IKafkaConsumerBuilder AddListener<T>(this IKafkaConsumerBuilder builder) where T : class, IKafkaConsumerListener
        {
            return builder.AddListener(typeof(T));
        }

        /// <summary>
        /// 创建客户端生产者
        /// </summary>
        /// <param name="kafkaProducerFactory"></param>
        /// <returns></returns>
        public static IClientProducer CreateClientProducer(this IKafkaProducerFactory kafkaProducerFactory)
        {
            return kafkaProducerFactory.CreateClientProducer(Options.DefaultName);
        }

        /// <summary>
        /// 发布消息
        /// </summary>
        /// <param name="producer"></param>
        /// <param name="message"></param>
        /// <param name="callback"></param>
        public static void Publish(this IClientProducer producer, object message, Action<DeliveryResult> callback = null)
        {
            producer.Publish(new object[] { message }, callback);
        }
        /// <summary>
        /// 发布消息
        /// </summary>
        /// <param name="producer"></param>
        /// <param name="messages"></param>
        /// <param name="callback"></param>
        public static void Publish(this IClientProducer producer, object[] messages, Action<DeliveryResult> callback = null)
        {
            producer.Publish(null, messages, callback);
        }
        /// <summary>
        /// 发布消息
        /// </summary>
        /// <param name="producer"></param>
        /// <param name="partition"></param>
        /// <param name="message"></param>
        /// <param name="callback"></param>
        public static void Publish(this IClientProducer producer, int? partition, object message, Action<DeliveryResult> callback = null)
        {
            producer.Publish(partition, new object[] { message }, callback);
        }
        /// <summary>
        /// 发布消息
        /// </summary>
        /// <param name="producer"></param>
        /// <param name="partition"></param>
        /// <param name="messages"></param>
        /// <param name="callback"></param>
        public static void Publish(this IClientProducer producer, int? partition, object[] messages, Action<DeliveryResult> callback = null)
        {
            producer.Publish(messages.Select(f => new KafkaMessage() { Partition = partition, Message = f }).ToArray(), callback);
        }
        /// <summary>
        /// 发布消息
        /// </summary>
        /// <param name="producer"></param>
        /// <param name="kafkaMessage"></param>
        /// <param name="callback"></param>
        public static void Publish(this IClientProducer producer, KafkaMessage kafkaMessage, Action<DeliveryResult> callback = null)
        {
            producer.Publish(new KafkaMessage[] { kafkaMessage }, callback);
        }
        /// <summary>
        /// 异步发布消息
        /// </summary>
        /// <param name="producer"></param>
        /// <param name="message"></param>
        /// <param name="callback"></param>
        public static async Task PublishAsync(this IClientProducer producer, object message, Action<DeliveryResult> callback = null)
        {
            await producer.PublishAsync(new object[] { message }, callback);
        }
        /// <summary>
        /// 异步发布消息
        /// </summary>
        /// <param name="producer"></param>
        /// <param name="messages"></param>
        /// <param name="callback"></param>
        public static async Task PublishAsync(this IClientProducer producer, object[] messages, Action<DeliveryResult> callback = null)
        {
            await producer.PublishAsync(null, messages, callback);
        }
        /// <summary>
        /// 异步发布消息
        /// </summary>
        /// <param name="producer"></param>
        /// <param name="partition"></param>
        /// <param name="message"></param>
        /// <param name="callback"></param>
        public static async Task PublishAsync(this IClientProducer producer, int? partition, object message, Action<DeliveryResult> callback = null)
        {
            await producer.PublishAsync(partition, new object[] { message }, callback);
        }
        /// <summary>
        /// 异步发布消息
        /// </summary>
        /// <param name="producer"></param>
        /// <param name="partition"></param>
        /// <param name="messages"></param>
        /// <param name="callback"></param>
        public static async Task PublishAsync(this IClientProducer producer, int? partition, object[] messages, Action<DeliveryResult> callback = null)
        {
            await producer.PublishAsync(messages.Select(f => new KafkaMessage() { Partition = partition, Message = f }).ToArray(), callback);
        }
        /// <summary>
        /// 异步发布消息
        /// </summary>
        /// <param name="producer"></param>
        /// <param name="kafkaMessage"></param>
        /// <param name="callback"></param>
        /// <returns></returns>
        public static async Task PublishAsync(this IClientProducer producer, KafkaMessage kafkaMessage, Action<DeliveryResult> callback = null)
        {
            await producer.PublishAsync(new KafkaMessage[] { kafkaMessage }, callback);
        }
        /// <summary>
        /// 异步发布消息
        /// </summary>
        /// <param name="producer"></param>
        /// <param name="kafkaMessages"></param>
        /// <param name="callback"></param>
        /// <returns></returns>
        public static async Task PublishAsync(this IClientProducer producer, KafkaMessage[] kafkaMessages, Action<DeliveryResult> callback = null)
        {
            await Task.Run(() =>
            {
                producer.Publish(kafkaMessages, callback);
            });
        }
    }
}
